package com.xujian.common.util;

import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.runtime.state.storage.FileSystemCheckpointStorage;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.io.IOException;
import java.util.concurrent.TimeUnit;

/**
 * @author star xu
 * @date 2022/12/29 17:49
 * @Description:
 * @Slogan: 致敬大师, 致敬未来的自己
 */
public class EnviromentUtil {
    public static StreamExecutionEnvironment build(String storgeType) throws IOException {
        //1.创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //2.flinkcdc 做断点续传,需要将flinkcdc读取binlog的位置信息以状态方式保存在checkpoint中即可.
        //参数设置
        //开启checkpoint 每隔5s 执行一次ck
        env.enableCheckpointing(5000);
        //高级选项：
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
        //设置模式为精确一次 (这是默认值)
        checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        //表示一旦Flink处理程序被cancel后，会保留CheckPoint数据，以便根据实际需要恢复到指定的CheckPoint。
        checkpointConfig.setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //设置失败重启策略，尝试重启3次，每隔120s重启1次
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, TimeUnit.SECONDS.toMillis(120)));
        //设置checkpoint的超时时间，即一次checkpoint必须在该时间内完成，默认为10分钟超时，不然就丢弃
        checkpointConfig.setCheckpointTimeout(600000);
        //设置两次checkpoint之间的最小时间间隔，防止出现例如状态数据过大而导致Checkpoint执行时间过长，从而导致Checkpoint积压过多
        // ，最终Flink应用密集地触发Checkpoint操作，会占用了大量计算资源而影响到整个应用的性能
        checkpointConfig.setMinPauseBetweenCheckpoints(500);
        //设置最大并行执行的检查点数量
        checkpointConfig.setMaxConcurrentCheckpoints(1);
        //容忍checkpoint失败多少次，默认0，不容忍
        checkpointConfig.setTolerableCheckpointFailureNumber(2);
        //设置存储备份
        if (StringUtils.isNotBlank(storgeType)){
            // 开启checkpoints的外部持久化
            // 目前代码不能设置保留的checkpoint个数 默认值时保留一个，假如要保留3个.可以在flink-conf.yaml中配置 state.checkpoints.num-retained: 3
            // 使用 RocksDBStateBackend 做为状态后端，enableIncrementalCheckpointing表示开启增量 Checkpoint
            EmbeddedRocksDBStateBackend stateBackend = new EmbeddedRocksDBStateBackend();
            env.setStateBackend(stateBackend);
            if ("obs".equalsIgnoreCase(storgeType)){
                checkpointConfig.setCheckpointStorage(new FileSystemCheckpointStorage("s3a://yjk-aimdt-dev/checkPoint/"));
            }else if ("oss".equalsIgnoreCase(storgeType)){
                checkpointConfig.setCheckpointStorage(new FileSystemCheckpointStorage("oss://yjk-aimdt-dev/checkPoint/"));
            }
        }
        // 有界数据流，则会采用批方式进行数据处理
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
        return env;
    }
}
